with  waybill_tmp as (
    select waybill_no,input_time,goods_type_code ,goods_type_name ,package_charge_weight, express_type_code , express_type_name
        from jms_dwd.dwd_yl_oms_oms_waybill_incre_dt
    where dt between date_add('{{ execution_date | cst_ds }}',-{{ params.out_days }}) and '{{ execution_date | cst_ds }}' and
date_format(input_time,'yyyy-MM-dd') >= date_add('{{ execution_date | cst_ds }}', -{{ params.out_days }})

),
shipment_tmp as (
      SELECT  shipment_no                   AS shipment_no             -- 转运任务单号
              ,network_code                 AS network_code            -- 网点编码
              ,min(roadsection_id)          AS roadsection_id          -- 路段编号
              ,min(planned_arrival_time)    AS planned_arrival_time    -- 计划到达日期+时间
              ,min(actual_arrival_time)     AS actual_arrival_time     -- 实际到达时间
              ,min(planned_departure_time)  AS planned_departure_time  -- 计划离开日期+时间
              ,min(actual_departure_time)   AS actual_departure_time   -- 实际发车时间
              ,min(latest_arrival_time)     AS latest_arrival_time     -- 最晚到车时间+日期
              FROM (
                    SELECT shipment_no, network_code, roadsection_id, planned_arrival_time,actual_arrival_time,planned_departure_time,actual_departure_time,null as latest_arrival_time FROM jms_ods.yl_tmsnew_branch_shipment_stop WHERE dt = '{{ execution_date | cst_ds }}'
                    AND (actual_arrival_time >= date_add('{{ execution_date | cst_ds }}', -{{ params.out_days }}) OR actual_departure_time >= date_add('{{ execution_date | cst_ds }}', -{{ params.out_days }}) OR
                    planned_arrival_time >= date_add('{{ execution_date | cst_ds }}', -{{ params.out_days }}) OR planned_departure_time >= date_add('{{ execution_date | cst_ds }}', -{{ params.out_days }}))
                    UNION ALL
                    SELECT shipment_no, network_code, roadsection_id, planned_arrival_time,actual_arrival_time,planned_departure_time,actual_departure_time,latest_arrival_time FROM jms_ods.yl_tmsnew_tms_shipment_stop
                    WHERE dt = '{{ execution_date | cst_ds }}'
                    AND (actual_arrival_time >= date_add('{{ execution_date | cst_ds }}', -{{ params.out_days }}) OR actual_departure_time >= date_add('{{ execution_date | cst_ds }}', -{{ params.out_days }}) OR
                    planned_arrival_time >= date_add('{{ execution_date | cst_ds }}', -{{ params.out_days }}) OR planned_departure_time >= date_add('{{ execution_date | cst_ds }}', -{{ params.out_days }}) OR
                    latest_arrival_time >= date_add('{{ execution_date | cst_ds }}', -{{ params.out_days }}))
                    ) as  stops GROUP BY shipment_no, network_code
           )

insert overwrite table jms_dwd.dwd_s01_whole_operations_trace_dt partition(dt)
select
trace_main.waybill_no
,trace_main.step
,trace_main.scan_type
,trace_main.network_code
,trace_main.network_type
,trace_main.network_name
,trace_main.network_scan_time_early
,trace_main.scan_time_early
,trace_main.scan_time_last
,trace_main.prev_network_code
,trace_main.prev_network_type
,trace_main.next_network_code
,trace_main.next_network_type
,trace_main.step_center
,trace_main.step_center_reserve
,trace_main.step_entrepot
,trace_main.step_entrepot_reserve
,trace_main.recordid
,trace_main.packagecode
,trace_main.nextstationcode
,trace_main.nextstation
,trace_main.destination
,trace_main.scanusercode
,trace_main.scanuser
,trace_main.source
,trace_main.weight
,trace_main.expresstype
,trace_main.returnflag
,trace_main.elescaleflag
,trace_main.pistolid
,waybill_tmp.goods_type_code
,waybill_tmp.goods_type_name
,waybill_tmp.package_charge_weight
,waybill_tmp.express_type_code
,waybill_tmp.express_type_name
,trace_main.shipment_no
,shipment_tmp.roadsection_id  as roadsection_id
,shipment_tmp.planned_arrival_time  as planned_arrival_time
,shipment_tmp.actual_arrival_time  as actual_arrival_time
,shipment_tmp.planned_departure_time  as planned_departure_time
,shipment_tmp.actual_departure_time  as actual_departure_time
,shipment_tmp.latest_arrival_time  as latest_arrival_time
,from_unixtime(unix_timestamp(),"yyyy-MM-dd HH:mm:ss") as update_time
,nvl(date_format(waybill_tmp.input_time,'yyyy-MM-dd'),'9999-99-99') as dt
 from (select * from jms_dwd.dwd_s01_whole_operations_trace_mid_dt where dt between date_add('{{ execution_date | cst_ds }}', -{{ params.out_days }}) and '{{ execution_date | cst_ds }}' ) trace_main
join waybill_tmp
     on trace_main.waybill_no = waybill_tmp.waybill_no
 left join  shipment_tmp
     on nvl(trace_main.shipment_no,trace_main.waybill_no) = shipment_tmp.shipment_no and nvl(trace_main.network_code,trace_main.waybill_no) = shipment_tmp.network_code
 distribute by dt, abs(hash(trace_main.waybill_no)) % 120 ;